package com.chenwc.util.multithread;

import com.alibaba.fastjson2.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Supplier;

/**
 * @author Calvin Chan
 */
public class SessionReferenceThreadAsync {
    private static final Logger logger = LoggerFactory.getLogger(SessionReferenceThreadAsync.class);

    /**
     * 异步执行，能获取执行结果
     */
    public static class GetResultCompletableFuture implements Supplier<JSONObject> {
        //计数器
        private final CountDownLatch doneSignal;
        //阻塞队列
        private BlockingQueue<MultiStepJob> queue = new LinkedBlockingQueue<>();
        //会话
        static Hashtable<MultiStepJob, MySessionReference> sessions = new Hashtable<>();
        //线程本地变量
        static ThreadLocal<MySessionReference> localSessionReference = new ThreadLocal<>();

        public GetResultCompletableFuture(CountDownLatch doneSignal, BlockingQueue<MultiStepJob> queue) {
            this.doneSignal = doneSignal;
            this.queue = queue;
        }

        @Override
        public JSONObject get() {
            JSONObject dataList = new JSONObject();
            try {
                //无限循环
                for (; ; ) {
                    //从队列中取出任务
                    MultiStepJob job = queue.poll(10, TimeUnit.SECONDS);
                    //队列为空，结束循环
                    if (job == null) {
                        return dataList;
                    }
                    MySessionReference sesRef = sessions.get(job);
                    if (sesRef == null) {
                        sesRef = new MySessionReference();
                        sessions.put(job, sesRef);
                    }
                    localSessionReference.set(sesRef);
                    logger.info("任务 " + job.getName() + " 开始。");
                    try {
                        JSONObject export = job.runNextStep();
                        if (export != null) {
                            for (String key : export.keySet()){
                                dataList.put(key, export.get(key));
                            }
                        }
                    } catch (Throwable th) {
                        th.printStackTrace();
                    }
                    if (job.isFinished()) {
                        logger.info("任务 " + job.getName() + " 结束。");
                        sessions.remove(job);
                        job.cleanUp();
                    } else {
                        logger.info("任务 " + job.getName() + " 激活。");
                        queue.add(job);
                    }
                    localSessionReference.set(null);
                }
            } catch (InterruptedException e) {
                //just leave
            } finally {
                doneSignal.countDown();
            }
            return dataList;
        }

    }

    public static class MySessionReference {

        //原子整数
        static AtomicInteger atomicInt = new AtomicInteger(0);
        //ID
        private final String id = "session-" + atomicInt.addAndGet(1);

        /**
         * 上下文已完成
         */
        public void contextFinished() {
            logger.debug("session " + id + " 上下文已完成");
        }

        /**
         * 上下文已开始
         */
        public void contextStarted() {
            logger.debug("session " + id + " 上下文已开始");
        }

        /**
         * 获取ID
         *
         * @return ID
         */
        public String getId() {
            return id;
        }
    }

    public static class MySessionReferenceProvider {

        /**
         * 获取当前会话参考
         *
         * @param scopeType 作用域类型
         * @return 会话参考
         */
        public MySessionReference getCurrentSessionReference(String scopeType) {
            //异步执行带返回结果的线程
            MySessionReference sesRef = GetResultCompletableFuture.localSessionReference.get();
            if (sesRef != null) {
                return sesRef;
            }
            throw new RuntimeException("未知线程:" + Thread.currentThread().getId());
        }

        /**
         * 会话是否存活
         *
         * @param sessionId 会话ID
         * @return 是否存活
         */
        public boolean isSessionAlive(String sessionId) {
            //异步执行带返回结果的线程
            Collection<MySessionReference> availableSessions = GetResultCompletableFuture.sessions.values();
            for (MySessionReference ref : availableSessions) {
                if (ref.getId().equals(sessionId)) {
                    return true;
                }
            }
            return false;
        }

        /**
         * Server会话继续
         *
         * @param sessionId 会话ID
         */
        public void serverSessionContinued(String sessionId) {
            logger.debug("Server会话继续, sessionID: " + sessionId);
        }

        /**
         * Server会话完成
         *
         * @param sessionId 会话ID
         */
        public void serverSessionFinished(String sessionId) {
            logger.debug("Server会话完成, sessionID: " + sessionId);
        }

        /**
         * Server会话激活
         *
         * @param sessionId 会话ID
         */
        public void serverSessionPassivated(String sessionId) {
            logger.debug("Server会话激活, sessionID: " + sessionId);
        }

        /**
         * Server会话开始
         *
         * @return 会话参考
         */
        public MySessionReference serverSessionStarted() {
            logger.debug("Server会话开始");
            return null;
        }
    }
}
